package org.hadoop.hive.Dao;

import org.hadoop.hive.Utils.Data;
import org.hadoop.hive.Utils.HbaseUtils;
import org.hadoop.hive.vo.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 作者：Lapper
 * QQ656277239
 * vx:JavaAdmin123
 */
@Repository
public class BaseDao extends Thread {
    @Autowired
    private HbaseUtils hbaseUtils;

    @Autowired
    private Data data;
    private  static String  name="bigdata";
    private  static  String   family="000001";

    private  static final String   rokey="rk00001";
    Integer allSalary = 0;
    Integer allage =0 ;
    Integer allSalary1 = 0;
    Integer allage1 =0 ;
    Integer allSalary2 = 0;
    Integer allage2 =0 ;
    String[] quailefierName={
            "Age",
            "Ebackground",
            "DateTime",
            "job",
            "major",
            "province",
            "salary"
    };
    public BaseDao() {
       Integer allSalary = null;
        Integer allage =null;
        Integer allSalary1 = null;
        Integer allage1 =null ;
        Integer allSalary2 = null;
        Integer allage2 =null ;
    }


    public  String BuildData() throws Exception {
        for (int i = 0; i < 100; i++) {
            int index = new Random().nextInt() * 100 + 1;
            String s = rokey + i + index;
            System.out.println("-----------------数据生成中---------------------" + i);
            hbaseUtils.Build(name, s, family, "Age", data.getAge());
            hbaseUtils.Build(name, s, family, "Ebackground", data.getEbackground());
            hbaseUtils.Build(name, s, family, "DateTime", data.getDateTime());
            hbaseUtils.Build(name, s, family, "job", data.getJobs());
            hbaseUtils.Build(name, s, family, "major", data.getmajor());
            hbaseUtils.Build(name, s, family, "province", data.getprovinceOrcity());
            hbaseUtils.Build(name, s, family, "salary", data.getSalary());

        }
                String ok = "数据更新成功";
                return ok;
    }
   public List<Alldata> alldata() throws Exception {
      Map<String,Object>     scan = hbaseUtils.scan("bigdata",0000000000000l ,9999999999999l );
      List<Alldata> list=new ArrayList<>();
          scan.forEach((k,v)->{
            Alldata alldata=new Alldata();
            alldata.setRowKey(k);
            alldata.setData(v);
            list.add(alldata);
          });
      List<Alldata> list1 = list.stream().limit(200l).collect(Collectors.toList());
       return list1;


   }
   public  List<GroupByAgeSalary> AgeAndSalary(){



       try {
//拿到全部
           Map<String,Object> map= hbaseUtils.scan("bigdata",1715434084572l ,9999999999999l );
           Collection<Object> collection =  map.values();
           //拿到所有的对象
           List<Map<String,Object>> list=new ArrayList<>();
           List<AgeAndSalary> allAgeAndSalary=new ArrayList<>();
           collection.forEach(s->{
               //对象强转为集合 装到list里面
               list.add((Map<String, Object>) s);
           });
           for (Map<String, Object> stringObjectMap : list) {
               //对数据循环拿到对应的年龄和薪资
               AgeAndSalary ageAndSalary=new AgeAndSalary();
               ageAndSalary.setSalary(Integer.parseInt(stringObjectMap.get("salary").toString()) );
               ageAndSalary.setAge(Integer.parseInt(stringObjectMap.get("Age").toString()));
               //拿到后装入新list
               allAgeAndSalary.add(ageAndSalary);
           }
           //准备三个list用来分组
           //青年人
           List  young=new ArrayList<AgeAndSalary>();
           //中年人
           List mid=new ArrayList<AgeAndSalary>();
           //老年人
           List old=new ArrayList<AgeAndSalary>();
           allAgeAndSalary.stream().forEach(s->{
               //分组方法
               if (s.getAge()>=18&s.getAge()<25){
                   //18-25岁
                   young.add(s);
               }
               if (s.getAge()>=25&s.getAge()<40){
                   //25-40岁
                   mid.add(s);
               }
               if (s.getAge()>=40&s.getAge()<55){
                   //40-65岁
                   old.add(s);
               }
           });
           //先准备一个List用来装最终的分组
           List<GroupByAgeSalary> list1=new ArrayList<>();
           //计算出青年的平均年龄以及平均薪资

           young.stream().forEach(
                   s-> {
                       //算出总年龄与总工资
                       AgeAndSalary ageAndSalary= (AgeAndSalary) s;
                       allage +=ageAndSalary.getAge();
                       allSalary+= ageAndSalary.getSalary();
                   });
           //算出总人数
           Long people=  young.stream().count();
           //算出平均工资
           Double AvgAge= (double) (allage / people);
           Double AvgSalary= (double) (allSalary /people);
           //封装装到vo容器里
           GroupByAgeSalary groupByAgeSalary=     new GroupByAgeSalary();
           groupByAgeSalary.setPeople("young");
           groupByAgeSalary.setAvgSalary(AvgSalary);
           groupByAgeSalary.setAvgAge(AvgAge);
           list1.add(groupByAgeSalary);
//计算中年人的平均年龄以及薪资
           mid.stream().forEach(
                   s-> {
                       //算出总年龄与总工资
                       AgeAndSalary ageAndSalary= (AgeAndSalary) s;
                       allage1 +=ageAndSalary.getAge();
                       allSalary1+= ageAndSalary.getSalary();
                   });
           //算出总人数
           Long people1= mid.stream().count();
           //算出平均工资
           Double AvgAge1= (double) (allage1 / people1);
           Double AvgSalary1= (double) (allSalary1 /people1);
           //封装装到vo容器里
           GroupByAgeSalary groupByAgeSalary1=     new GroupByAgeSalary();
           groupByAgeSalary1.setPeople("mid");
           groupByAgeSalary1.setAvgSalary(AvgSalary1);
           groupByAgeSalary1.setAvgAge(AvgAge1);
           list1.add(groupByAgeSalary1);
           //计算老年人
           old.stream().forEach(
                   s-> {
                       //算出总年龄与总工资
                       AgeAndSalary ageAndSalary= (AgeAndSalary) s;
                       allage2 +=ageAndSalary.getAge();
                       allSalary2+= ageAndSalary.getSalary();
                   });
           //算出总人数
           Long people2= mid.stream().count();
           //算出平均工资
           Double AvgAge2= (double) (allage2 / people2);
       Double AvgSalary2= (double) (allSalary2 /people2);
           //封装装到vo容器里
           GroupByAgeSalary groupByAgeSalary2=     new GroupByAgeSalary();
           groupByAgeSalary2.setPeople("old");
           groupByAgeSalary2.setAvgSalary(AvgSalary2);
           groupByAgeSalary2.setAvgAge(AvgAge2);
           list1.add(groupByAgeSalary2);
           return list1;

       } catch (Exception e)
       {
           throw new RuntimeException(e);
       }finally {
          allSalary = 0;
          allage =0;
           allSalary1 = 0;
          allage1 =0;
          allSalary2 = 0;
            allage2 =0;
       }

   }
   public Map<String,Integer> Site() throws Exception {
       Map<String, Object> scan = hbaseUtils.scan("bigdata", 1715000613827l, 9999999999999l);
       //拿到所有值
       Collection collection=scan.values();
       //准备两个数组 一个存放强制转换的对象 另外一个存放最终要的地理位置信息
       List<Map<String,Object>> results=new ArrayList<>();
       List<HotProvince> endresult=new ArrayList<>();
       collection.stream().forEach(
               s->{
                   //强制转换为对象 并且存到list里面
                   results.add((Map<String, Object>) s);
               }
       );
      results.stream().forEach(s->{
          HotProvince hotProvince=new HotProvince();
          hotProvince.setProvince(s.get("province").toString());
          hotProvince.setCount(1);
          endresult.add(hotProvince);
      });

      Map<String,Integer> map=new HashMap<>();
      endresult.forEach(s->{
          if (map.containsKey(s.getProvince()))
          {
              map.replace(s.getProvince(),map.get(s.getProvince())+1);
          }else {
              map.put(s.getProvince(),s.getCount());
          }
      });
//计数
       System.out.println(map);


       return map;
   }
   public Map<String, Integer> majorAndJobs() throws Exception {
       Map<String, Object> scan = hbaseUtils.scan("bigdata", 0000000000000l, 9999999999999l);
       //拿到所有值
       Collection collection=scan.values();
       //先对collection进行循环  把Object强转为Map
       List<Map<String,Object>> results=new ArrayList<>();
       List<Majorandjob> majorandjobs=new ArrayList<>();
       collection.stream().forEach(s->{
           results.add((Map<String, Object>) s);
       });
       //装填完毕
       results.stream().forEach(s->{
           Majorandjob majorandjob=new Majorandjob();

           majorandjob.setJob(s.get("job").toString());
           majorandjob.setMajor(s.get("major").toString());
           majorandjob.setMajors(1);
           majorandjob.setJobs(1);
           majorandjobs.add(majorandjob);
       });
       //重复的计数
       Map<String,Integer> map=new HashMap<>();
       majorandjobs.stream().forEach(s->{
           if (map.containsKey(s.getMajor())){
               map.replace(s.getMajor(),map.get(s.getMajor())+1);
           }else {
               map.put(s.getMajor(),s.getMajors());
           }
           if (map.containsKey(s.getJob())){
               map.replace(s.getJob(),map.get(s.getJob())+1);
           }else {
               map.put(s.getJob(),s.getJobs());
           }
       });
       return  map;

   }

    public String remove(String rowKey) throws Exception {
        for (String s : quailefierName) {
            hbaseUtils.delete(name,rowKey,family,s);
        }
       return "ok";
    }

    public Map<String, String> Select(String rowKey) throws Exception {
        Map<String, String> map = hbaseUtils.get(name, rowKey);
        return map;
    }
    public Map<String,Integer> EbackgroundandJobs() throws Exception {
        Map<String, Object> scan = hbaseUtils.scan("bigdata", 0000000000000l, 9999999999999l);
        Collection<Object> values = scan.values();
        List<Map<String,Object>> results=new ArrayList<>();
List<Ebackgroundandjobs> list=new ArrayList<>();
      //先拿到所有对象

        for (Object value : values) {
            results.add((Map<String, Object>) value);
        }
        results.stream().forEach(s->{
            Ebackgroundandjobs ebackgroundandjobs=new Ebackgroundandjobs();
            ebackgroundandjobs.setEbackground(s.get("Ebackground").toString());
            ebackgroundandjobs.setJobs(1);
            list.add(ebackgroundandjobs);
        });
Map<String,Integer> result=new HashMap<>();
        list.stream().forEach(s->{
            if (result.containsKey(s.getEbackground())){
                result.replace(s.getEbackground(),result.get(s.getEbackground())+1);
            }else {
                result.put(s.getEbackground(),s.getJobs());
            }
        });

        return result ;
    };
}
